package forward

import (
	"github.com/gorilla/websocket"
	"k8s.io/klog"
	"time"
)

// CenterHandler 全局处理中心，关联着每个 Client 的注册、注销、广播通道，相当于每个用户的中心通讯的中介。
type CenterHandler struct {
	// 全局广播
	GlobalBroadcast chan []byte
	// 分组广播 map的key使用业务id 与用户集合中的value相对应
	PacketBroadcast chan map[string][]byte
	// 注册通道，有用户进来 则推到用户集合map中
	Register chan *Client
	// 注销通道，有用户关闭连接 则将该用户剔出集合map中
	Logout chan *Client
	// 正在使用的用户集合 key是客户端指针 value是业务id
	Clients map[*Client]string
}

// Monitoring 处理中心的一个接口，监控状态
func (ch *CenterHandler) Monitoring() {
	for {
		select {
		case client := <-ch.Register:
			// 注册，新用户连接过来会推进注册通道
			ch.Clients[client] = client.BusinessId
			// 注销，关闭连接或连接异常会将用户推出群聊
		case client := <-ch.Logout:
			delete(ch.Clients, client)
			// 监听全局广播
		case message := <-ch.GlobalBroadcast:
			for client := range ch.Clients {
				if "pong" != string(message) {
					client.Send <- message
					klog.Info("【全局广播】的信息--->" + string(message))
				} else {
					if time.Now().After(client.Overtime.Add(5 * time.Second)) {
						klog.Error("响应超时，全局已断开连接")
						go client.StopHandler()
					}
				}
			}
			// 监听分组广播
		case msgMap := <-ch.PacketBroadcast:
			for businessId, message := range msgMap {
				for client, business := range ch.Clients {
					if businessId == business {
						if "pong" != string(message) {
							client.Send <- message
							klog.Info("【" + business + "】组内广播的信息--->" + string(message))
						} else {
							if time.Now().After(client.Overtime.Add(5 * time.Second)) {
								klog.Error("响应超时，【" + business + "】组已断开连接")
								go client.StopHandler()
							}
						}
					}
				}
			}
		}
	}
}

// BroadcastType 广播类型
type BroadcastType string

const (
	Global BroadcastType = "Global"
	Packet BroadcastType = "Packet"
)

// Client 抽象出来的 Client，里面有这个 websocket 连接的 读 和 写 操作
type Client struct {
	BusinessId string
	Handler    *CenterHandler
	Conn       *websocket.Conn
	Send       chan []byte
	DoCustomEventClient
	Exception bool
	// 启用的广播类型 全局广播/分组广播
	SelectBroadcast BroadcastType
	// 超时时间
	Overtime time.Time
}

// DoCustomEventClient 自定义的事件处理任务
type DoCustomEventClient interface {
	DoCloseEvent(c *Client)
	DoExitEvent(c *Client)
}

// StopHandler 断开连接
func (c *Client) StopHandler() {
	c.Handler.Logout <- c
	c.Conn.Close()
}

// CloseHandler 关闭连接
func (c *Client) CloseHandler() {
	if c.Exception {
		c.DoExitEvent(c)
	} else {
		c.DoCloseEvent(c)
	}
	c.Handler.Logout <- c
	c.Conn.Close()
}

// WriteHandler 写，主动推送消息给客户端
func (c *Client) WriteHandler() {
	for {
		// 广播推过来的新消息，马上通过websocket推给自己
		message, _ := <-c.Send
		if err := c.Conn.WriteMessage(websocket.TextMessage, message); nil != err {
			klog.Error(err)
			return
		}
	}
}

// ReadHandler 读，监听客户端是否有推送内容过来服务端
func (c *Client) ReadHandler() {
	for {
		// 循环监听该用户是否要发言
		_, message, err := c.Conn.ReadMessage()
		if nil != err {
			// 异常关闭的处理
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway,
				websocket.CloseAbnormalClosure) {
				c.Exception = true
			}
			c.CloseHandler()
			return
		}
		// 推给广播中心，广播中心再推给每个用户
		switch c.SelectBroadcast {
		case Global:
			c.Handler.GlobalBroadcast <- message
		case Packet:
			data := make(map[string][]byte)
			data[c.BusinessId] = message
			c.Handler.PacketBroadcast <- data
		}
	}
}

// Ping 探测
func (c *Client) Ping() {
	for {
		if err := c.Conn.WriteMessage(websocket.TextMessage, []byte("ping")); nil != err {
			// 异常关闭的处理
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway,
				websocket.CloseAbnormalClosure) {
				c.Exception = true
			}
			c.CloseHandler()
			return
		}
		c.Overtime = time.Now()
		time.Sleep(10 * time.Second)
	}
}
